import java.net.URI

import org.apache.spark.{SparkConf, SparkContext}

object Q3_click_imp {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setAppName(this.getClass.getCanonicalName.init)//.setMaster("local[*]")
    val sc = new SparkContext(conf)
    sc.setLogLevel("WARN")

    //读文件
/*
    val click = sc.textFile("F:\\TrainingCourse\\P1 大数据\\大数据正式班第四阶段模块二\\spark实战应用（上）\\代码\\data\\click.log")
    val imp = sc.textFile("F:\\TrainingCourse\\P1 大数据\\大数据正式班第四阶段模块二\\spark实战应用（上）\\代码\\data\\imp.log")*/
    val click = sc.textFile(args(0))
    val imp = sc.textFile(args(1))
    //获取字段adid
    val clickCount = click.flatMap(s => uri2adid(s.split(" ").last)).map((_,1)).reduceByKey(_ + _)
    val impCount = imp.flatMap(s => uri2adid(s.split(" ").last)).map((_,1)).reduceByKey(_ + _)
    clickCount
      .fullOuterJoin(impCount)
      .saveAsTextFile("/result/result.rdd")
    //
    sc.stop()

  }
  def uri2adid(s:String):Array[Long] = {
    new URI(s).getQuery.split("&").filter(_.matches("adid=[\\d]+")).map(_.split("=")(1).toLong)
  }
}